package com.dys.flink.wc.netty.handler;

import com.dys.flink.wc.model.Product;
import com.dys.flink.wc.util.JsonUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;

public class ProductEncoder extends MessageToByteEncoder<Product> {
    private static final Logger logger = LoggerFactory.getLogger(ProductEncoder.class);
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    @Override
    protected void encode(ChannelHandlerContext ctx, Product product, ByteBuf out) throws Exception {
        try {
            // 使用 Jackson 将对象转换为 JSON 字符串
            String json = JsonUtils.toJson(product);
            byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
            
            // 添加长度前缀解决粘包问题
            out.writeInt(bytes.length);
            out.writeBytes(bytes);
            
            logger.debug("Successfully encoded product: {}", product);
        } catch (Exception e) {
            logger.error("Failed to encode product: {}", product, e);
            throw e;
        }
    }
}